package com.shujia.spark.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}

object Demo5Window {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setAppName("stateByKeu")
    conf.setMaster("local[2]")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val wordsDS: DStream[String] = linesDS.flatMap(_.split("<"))

    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))

    /**
      * 统计单词的数量，统计最近15秒(窗口大小)，每隔5秒（滑动时间）统计一次
      *
      */

    /*  val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
        (x: Int, y: Int) => x + y, //聚合函数
        Durations.seconds(15), //窗口大小
        Durations.seconds(5) //滑动时间
      )*/

    //优化版本
    val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x: Int, y: Int) => x + y, //聚合函数
      (i: Int, j: Int) => i - j, //减去多余数据的函数
      Durations.seconds(15), //窗口大小
      Durations.seconds(5) //滑动时间
    )

    countDS.filter(_._2 != 0).print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
